package com.mlming.simpledb.backend.dm.logger;

import com.google.common.primitives.Bytes;
import com.mlming.simpledb.backend.utils.Panic;
import com.mlming.simpledb.backend.utils.Parser;

import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.util.Arrays;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import com.mlming.simpledb.common.Error;
import sun.rmi.runtime.Log;

/**
 * 日志文件读写
 *
 * 日志文件标准格式为：
 * [XChecksum] [Log1] [Log2] ... [LogN] [BadTail]
 * XChecksum 为后续所有日志计算的Checksum，int类型
 *
 * 每条正确日志的格式为：
 * [Size] [Checksum] [Data]
 * Size 4字节int 标识Data长度
 * Checksum 4字节int
 */
public class LoggerImpl implements Logger {

    private static final int SEED = 13331; // 用于计算校验和的一个参数 => 校验和的计算规则: 每次*SEED + 当前byte

    private static final int OF_LOGS = 4; // 整个日志文件开头是4字节的XChecksum, 则4字节就是后续日志记录的偏移量
    private static final int OF_SIZE = 0; // 每一条日志的size字段的偏移量
    private static final int OF_CHECKSUM = OF_SIZE + 4; // 每一条日志的Checksum字段的偏移量
    private static final int OF_DATA = OF_CHECKSUM + 4; // 每一条日志的data部分的偏移量

    public static final String LOG_SUFFIX = ".log"; // 日志文件的后缀名 .log

    private RandomAccessFile LogFile; // 代表着日志文件
    private FileChannel fc;// 读写日志文件类
    private Lock lock; // 锁

    private long position;  // 当前日志文件的指针的位置 => 一般指向某条记录的开始位置
    private long fileSize;  // 初始化时记录，log操作不更新
    private int xChecksum; // 当前整个日志文件的校验和

    public LoggerImpl(RandomAccessFile raf, FileChannel fc) {
        this.LogFile = raf;
        this.fc = fc;
        lock = new ReentrantLock();
    }

    public LoggerImpl(RandomAccessFile raf, FileChannel fc, int xChecksum) {
        this.LogFile = raf;
        this.fc = fc;
        this.xChecksum = xChecksum;
        lock = new ReentrantLock();
    }

    // 通过日志文件, 初始化fileSize 与 xChecksum
    void init() {
        long size = 0;
        try {
            size = LogFile.length();
        } catch (IOException e) {
            Panic.panic(e);
        }
        if(size < 4) {
            Panic.panic(Error.BadLogFileException);
        }
        ByteBuffer raw = ByteBuffer.allocate(4);
        try {
            fc.position(0);
            fc.read(raw);
        } catch (IOException e) {
            Panic.panic(e);
        }
        int xChecksum = Parser.parseInt(raw.array());
        // 赋值
        this.fileSize = size;
        this.xChecksum = xChecksum;
        // 检查完整性并且截断不完整的日志
        checkAndRemoveTail();
    }
    // 定位到整个日志文件的第一条记录开始处
    @Override
    public void rewind() {
        position = OF_LOGS;
    }
    @Override
    // 从日志文件的x位置开始截断
    public void truncate(long x) throws Exception {
        lock.lock();
        try {
            fc.truncate(x);
        } finally {
            lock.unlock();
        }
    }

    // 通过读取整个日志文件的一条条日志, 来计算整个日志的校验和, 从而判断完整性,
    // 如果发现不完整,应该报错
    // 最后要把该日志最后指针读取位置(要么就是不完整的日志 , 要么就是最后一行)的后面进行一个截断,确保没有最后没有不完整的日志记录
    private void checkAndRemoveTail() {
        // 1) 先定位到整个日志文件的第一条记录的开始处
        rewind();
        // 2) 一条条读取日志记录,计算整个日志的校验和, 直到读取到不完整的日志或者最后一行(internNext()返回null)
        int xCheckSum = 0;
        byte[] log = null;
        while ((log = internNext()) != null) {
            xCheckSum = calChecksum(xCheckSum,log);
        }
        // 3) 判断是否和之前就赋值过的成员变量是否一致,如果不一致,直接报错
        if(xCheckSum != this.xChecksum) {
            Panic.panic(Error.BadLogFileException);
        }
        // 4) 把该日志最后指针读取位置(要么就是不完整的日志 , 要么就是最后一行)的后面进行一个截断,确保没有最后没有不完整的日志记录
        try {
            truncate(position);
        } catch (Exception e) {
            Panic.panic(e);
        }
        try {
            LogFile.seek(position);
        } catch (IOException e) {
            Panic.panic(e);
        }
        // 5) 检查完毕, 要把指针回到第一条记录开始位置
        rewind();
    }

    /**
     * 计算出当前的校验和 => 既可以用于计算整个日志文件的校验和XCheckSum, 又可以用于计算单条日志记录的校验和checkSum
     * @param xCheck 如果是XCheckSum,则该值是当前整个日志文件的XCheckSum ; 如果是checkSum, 则是0
     * @param log 如果是XCheckSum,则该值是最新加入的那一整条日志记录(包括size,checkSum,data); 如果是checkSum, 则是当前日志记录中的data域
     * @return 返回 加上了最新这条日志记录的XCheckSum  或者  当前传入的这个日志记录的data对应的checkSum
     */
    private int calChecksum(int xCheck, byte[] log) {
        // 计算规则: 每次*SEED 再加 当前字节
        for (byte b : log) {
            xCheck = xCheck * SEED + b;
        }
        return xCheck;
    }

    // 读取一条日志记录的data域, 同时指针移动到下一条日志开始位置(类似于迭代器)
    // 规则: 如果发现这条日志记录不完整: 1) size相关判断 2) checkSum相关判断
    //      则直接返回null, 使其调用者得知当前日志记录不完整
    //      之所以返回null, 是因为可以和"读取下一条"的逻辑相适应:
    //          因为当我们读取下一条发现没有记录时, 肯定也是返回null来告知调用者日志已经读取到最后一行了
    //      而对于一个日志而言, 当出现日志不完整的情况, 一般就是数据库异常关闭的情况, 那么就可以认为不完整的这一记录, 就是最后一条记录
    //      所以, 都返回null, 逻辑相统一
    private byte[] internNext() {
        // 注意: position代表着当前指针位置, 一般指向某个日志的开始位置
        // 1) 如果当前指针位置+size字段+checkSum字段已经大于等于文件大小了, 说明日志记录不完整/到了最后一行
        if(position + OF_DATA >=  fileSize) {
            return null;
        }
        // 2) 读取出当前日志的size字段 => data域的字节长度
        ByteBuffer buf = ByteBuffer.allocate(OF_CHECKSUM);
        try {
            fc.position(position); // 定位到日志记录的开始位置
            fc.read(buf); // 读取到buf中
        } catch (IOException e) {
            Panic.panic(e);
        }
        int size = Parser.parseInt(buf.array());
        // 如果当前指针位置+size字段+checkSum字段+data占用空间已经大于文件大小了, 说明日志记录不完整
        if(position + OF_DATA + size >  fileSize) {
            return null;
        }
        // 3) 读取出当前一整条日志记录
        ByteBuffer tem = ByteBuffer.allocate(OF_DATA + size);
        try {
            fc.position(position); // 定位到日志记录的开始位置
            fc.read(tem); // 读取到buf中
        } catch (IOException e) {
            Panic.panic(e);
        }
        byte[] log = tem.array();
        // 4) 从log中, 读取出校验和字段 和 data部分, 通过data部分计算校验和, 看看是否与读取的校验和一致
        int checkSumRead = Parser.parseInt(Arrays.copyOfRange(log, OF_CHECKSUM, OF_DATA));
        int checkSumCompute = calChecksum(0,Arrays.copyOfRange(log, OF_DATA, log.length));
        // 如果读取的校验和 和 计算出的校验和不一致, 说明日志不完整
        if(checkSumRead != checkSumCompute) {
            return null;
        }
        // 5) 把指针移动到下一条记录的位置:
        position += log.length;
        // 返回当前日志记录log
        return log;
    }

    @Override
    // 读取一条日志记录的data域, 同时指针移动到下一条日志开始位置(类似于迭代器)
    // 基于internNext方法读取出完整的日志记录来实现取出data域
    public byte[] next() {
        lock.lock(); // 因为要读取共享资源, 所以要加锁
        try {
            byte[] log = internNext(); // 通过internNext方法读取一条完整日志记录,同时指针移动到下一条日志开始位置
            if(log == null) return null; // 如果是null, 说明日志到最后一行或者是不完整, 自然没有data域, 则也返回null
            return Arrays.copyOfRange(log,OF_DATA,log.length); // 返回data域
        } finally {
            lock.unlock();
        }
    }

    @Override
    public void close() {
        try {
            fc.close();
            LogFile.close();
        } catch(IOException e) {
            Panic.panic(e);
        }
    }

    @Override
    // 追加一条日志记录到日志文件中
    public void log(byte[] data) {
        // 1) 先把byte[] data包裹为我们的单条日志格式[size][checkSum][data]
        byte[] log = wrapLog(data);
        // 2) 把该条日志记录追加到日志文件中去:
        ByteBuffer buf = ByteBuffer.wrap(log);
        lock.lock();
        try {
            fc.position(fc.size()); // 定位到文件末尾
            fc.write(buf); // 写入文件中
            // 3) 更新整个日志文件的校验和
            updateXCheckSum(log);
        } catch (IOException e) {
            Panic.panic(e);
        } finally {
            lock.unlock();
        }
    }
    // 更新整个日志文件的校验和
    private void updateXCheckSum(byte[] log) {
        this.xChecksum = calChecksum(this.xChecksum, log); // 在当前xChecksum,再追加log计算
        try {
            fc.position(0);
            fc.write(ByteBuffer.wrap(Parser.int2Byte(xChecksum)));
            fc.force(false);
        } catch(IOException e) {
            Panic.panic(e);
        }
    }

    // 把data域包裹为我们的单条日志格式[size][checkSum][data]
    private byte[] wrapLog(byte[] data) {
        byte[] size = Parser.int2Byte(data.length);
        byte[] checkSum = Parser.int2Byte(calChecksum(0,data));
        return Bytes.concat(size,checkSum,data); // 拼接三个byte数组
    }
}
